﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;

namespace Bohdan.Arch.Messaging
{
    public abstract class GeneralSubscriber : IDisposable
    {
        protected Semaphore _sem = new Semaphore(0, int.MaxValue);
        protected SortedSet<GeneralMessage> _queue = new SortedSet<GeneralMessage>();
        protected EventWaitHandle _ewhExit = new EventWaitHandle(false, EventResetMode.ManualReset);
        protected volatile bool _disposed = false;

        public abstract ThreadPriority WorkerThreadPriority { get; }
        public abstract Guid SubscriberId { get; }
        protected abstract void ProcessMessage(GeneralMessage msg);
        public virtual MessagingBus Bus { get; set; }

        public abstract void StartWorkersSync();
        public abstract void EndWorkersSync();

        public GeneralSubscriber()
        {
            StartWorkersSync();
        }

        protected virtual void ProcessMessageImmediateCaller(GeneralMessage msg)
        {
            ProcessMessage(msg);
        }

        protected virtual void ResortOnReporting(GeneralMessage msg)
        {
            throw new NotImplementedException(string.Format("Handling of message of type {0} is not implemented in class {1}.", msg.GetType().Name, this.GetType().Name));
        }

        protected virtual void ProcessMessageCaller(GeneralMessage msg)
        {
            try
            {
                msg.StartedProcessing = DateTime.Now;
                if (msg.PreExecutionValidate())
                {
                    try
                    {
                        msg.ReportBeingProcessed();
                        msg.BeforeExecutionFire();
                        ProcessMessageImmediateCaller(msg);
                    }
                    catch (Exception ex)
                    {
                        Bus.FireExceptionHandler(this, new MessagingException("Error processing message " + msg.ToString(), ex));
                    }
                    finally
                    {
                        msg.FinishedProcessing = DateTime.Now;
                        msg.AfterExecutionFire();
                    }
                }
                else
                    msg.FinishedProcessing = DateTime.Now;
            }
            finally
            {
                msg.DesclareMessageProcessed();
            }
        }
        protected virtual void PopAndProcessMessage()
        {
            // get message
            GeneralMessage msg = null;
            lock (_queue)
            {
                msg = _queue.Max;
                _queue.Remove(msg);
            }

            ProcessMessageCaller(msg);
        }
        protected virtual void WorkerThread()
        {
#if TRACE_MESSAGE_BUS
            Trace.TraceInformation(GetType().Name + " worker thread started.");
#endif

            WaitHandle[] wh = { _ewhExit, _sem };
            while (WaitHandle.WaitAny(wh) != 0)
                PopAndProcessMessage();

#if TRACE_MESSAGES
            Trace.TraceInformation(GetType().Name + " worker thread exited.");
#endif
        }

        public virtual void AddMessageAsync(GeneralMessage msg)
        {
            lock (this)
            { 
                if (_disposed)
                    throw new ObjectDisposedException(this.GetType().Name + " is disposed.");
                lock (_queue)
                {
                    _queue.Add(msg);
                    _sem.Release();
                }
            }
        }
        public virtual void Dispose()
        {
            Trace.TraceInformation("Dispose of {0} requested.", this.GetType().Name);
            lock (this)
            {
                if (_disposed)
                    return;
                _ewhExit.Set();
                EndWorkersSync();
                Trace.TraceInformation("Dispose of {0} done.", this.GetType().Name);
                _disposed = true;
            }
        }
    }
}
